package com.waverider.promotion.biz.task;

import net.sf.json.JSONObject;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;

import com.iacrqq.util.StringUtil;
import com.taobao.api.internal.stream.Configuration;
import com.taobao.api.internal.stream.TopCometStream;
import com.taobao.api.internal.stream.TopCometStreamFactory;
import com.taobao.api.internal.stream.connect.ConnectionLifeCycleListener;
import com.taobao.api.internal.stream.message.TopCometMessageListener;
import com.waverider.promotion.biz.manager.OrderManager;
import com.waverider.promotion.constants.PromotionConstants;

/**
 * 
 * @author sihai
 *
 */
public class MessageReceiver implements InitializingBean
{
	
	private static final Log logger = LogFactory.getLog(MessageReceiver.class);
	
	private Configuration configuration;
	private TopCometStream stream;
	
	private OrderManager orderManager;
	
	private MessageConsumer messageConsumer;
	
	public void setOrderManager(OrderManager orderManager)
	{
		this.orderManager = orderManager;
	}


	private void _init_()
	{	
		messageConsumer = new MessageConsumer();
		configuration = new Configuration(PromotionConstants.BG_APP_KEY,PromotionConstants.APP_SECRET,null);
		
		configuration.setMinThreads(1);			//处理消息的线程池中线程最小个数
		configuration.setMaxThreads(8);			//处理消息的线程池中线程最大个数
		configuration.setQueueSize(1024);		//处理消息的线程池中队列大小
		stream = new TopCometStreamFactory(configuration).getInstance();
		stream.setConnectionListener(new ConnectionListener());
		stream.setMessageListener(new TopCometMessageListenerImpl());
		stream.start();
		
		logger.info("MessageReceiver initizied");
	}
	
	
	@Override
	public void afterPropertiesSet() throws Exception
	{
		_init_();
	}
	
	class ConnectionListener implements ConnectionLifeCycleListener{
		public void onConnect() {
			logger.info("connect to stream success");
		}

		public void onDisconnect() {
			logger.warn("stream disconnect...");
		}

		public void onException(Throwable t) {
			logger.error(t);
		}

		public void onConnectError(Exception e) {
			logger.error("connect to stream error:" + e);
		}

		public void onReadTimeout() {
			logger.error("read stream timeout");
		}

		public void onSysErrorException(Exception e) {
			logger.error("stream system error:" + e);
		}

		public void onReconnect() {
			logger.warn("reconnect to stream");
		}

		public void onMaxReadTimeoutException() {
			logger.error("stream maxReadTimeoutException");
		}
	}
	
	class TopCometMessageListenerImpl implements TopCometMessageListener{

		public void onReceiveMsg(String message) {
			System.out.println("receive:" + message);
			messageConsumer.consume(message);
		}

		public void onConnectReachMaxTime() {
			System.out.println("reach max time");
		}

		public void onDiscardMsg(String message) {
			System.out.println("discardMsg:" + message);
		}

		public void onServerUpgrade(String message) {
			System.out.println("DeployMsg:" + message);
		}

		public void onServerRehash() {
			System.out.println("server rehash");
		}

		public void onServerKickOff() {
			System.out.println("server kickoff");
		}

		public void onOtherMsg(String message) {
			System.out.println("otherMsg:" + message);
		}

		public void onException(Exception ex) {
			System.out.println("exception:" + ex);
		}

		public void onConnectMsg(String message) {
			System.out.println("connection success message:" + message);	
		}

		public void onHeartBeat() {
			System.out.println("heart beat");
		}

		public void onClientKickOff() {
			System.out.println("client kick off");
		}
	}
	
	private class MessageConsumer
	{
		private static final String MSG_TOPIC = "topic";
		private static final String MSG_TOPIC_TRADE = "trade";
		private static final String MSG_TOPIC_STATUS = "trade";
		private static final String MSG_STATUS_TRADE_CREATE = "TradeCreate";
		private static final String MSG_STATUS_TRADE_MODIFY_FEE = "TradeModifyFee";
		private static final String MSG_STATUS_TRADE_CLOSE_AND_MODIFY_DETAIL_ORDER = "TradeCloseAndModifyDetailOrder";
		private static final String MSG_STATUS_TRADE_CLOSE = "TradeClose";
		private static final String MSG_STATUS_TRADE_BUYER_PAY = "TradeBuyerPay";
		private static final String MSG_STATUS_TRADE_SELLER_SHIP = "TradeSellerShip";
		private static final String MSG_STATUS_TRADE_SUCCESS = "TradeSuccess";
		
		
		private void consume(String msg)
		{
			JSONObject json = JSONObject.fromObject(msg);
			if(StringUtil.equals(json.getString(MSG_TOPIC), MSG_TOPIC_TRADE))
			{
				
				if(StringUtil.equals(json.getString(MSG_TOPIC_STATUS), MSG_STATUS_TRADE_CREATE))
				{
					// 创建交易
					
				}
				else if(StringUtil.equals(json.getString(MSG_TOPIC_STATUS), MSG_STATUS_TRADE_BUYER_PAY))
				{
					// 付款
					orderManager.syncTrade(json.getLong("user_id"), json.getLong("tid"));
				}
				else if(StringUtil.equals(json.getString(MSG_TOPIC_STATUS), MSG_STATUS_TRADE_SUCCESS))
				{
					// 交易成功
					orderManager.syncTrade(json.getLong("user_id"), json.getLong("tid"));
				}
			}
			else
			{
				logger.warn(new StringBuilder("Not trade message, discard message:").append(msg));
			}
		}
	}
}
